package publisher;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.ConfirmListener;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;

import java.io.IOException;
import java.util.Collections;
import java.util.SortedSet;
import java.util.TreeSet;
import java.util.concurrent.TimeoutException;

/**
 * Created with IntelliJ IDEA.
 * Description:
 * User: hp
 * Date: 2025-04-01
 * Time: 21:14
 */
public class PublisherConfirms {
    static Connection createConnection() throws IOException, TimeoutException {
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost("1.95.194.117");
        connectionFactory.setPort(5672);
        connectionFactory.setUsername("wangkun");
        connectionFactory.setPassword("200411");
        connectionFactory.setVirtualHost("study");
        Connection connection = connectionFactory.newConnection();

        return connection;
    }

    public static void main(String[] args) throws IOException, InterruptedException, TimeoutException {
        //单独确认
        publishingMessagesIndividually();

        //批量确认
        publishingMessagesInBatches();

        //异步确认
        handlingpublisherConfirmsAsynchronously();


    }

    private static void handlingpublisherConfirmsAsynchronously() throws IOException, TimeoutException, InterruptedException {
        try (Connection connection = createConnection()) {
            //1.开启信道
            Channel channel = connection.createChannel();
            //2.设置Confirm模式
            channel.confirmSelect();
            //3.声明队列
            channel.queueDeclare("queue3",true,false,false,null);
            //4.监听Confirm
            long start = System.currentTimeMillis();
            SortedSet<Long> confirmSeqNo = Collections.synchronizedSortedSet(new TreeSet<>());
            channel.addConfirmListener(new ConfirmListener() {
                @Override
                public void handleAck(long deliveryTag, boolean multiple) throws IOException {
                    if (multiple) {
                        confirmSeqNo.headSet(deliveryTag + 1).clear();
                    }else {
                        confirmSeqNo.remove(deliveryTag);
                    }
                }

                @Override
                public void handleNack(long deliveryTag, boolean multiple) throws IOException {
                    if (multiple) {
                        confirmSeqNo.headSet(deliveryTag + 1).clear();
                    }else {
                        confirmSeqNo.remove(deliveryTag);
                    }
                }
            });

            //5.发送消息
            for (int i = 0; i < 1000; i++) {
                String message = "hello publisher Confirms  " +  i;
                long seqNo = channel.getNextPublishSeqNo();
                channel.basicPublish("","queue3",null,message.getBytes());
                confirmSeqNo.add(seqNo);
            }

            while (!confirmSeqNo.isEmpty()) {
                Thread.sleep(10);
            }
            long end = System.currentTimeMillis();
            System.out.println("异步确认策略消息条数: " + 1000 + " 耗时: " + (end-start));
        }
    }

    private static void publishingMessagesInBatches() throws IOException, TimeoutException, InterruptedException {
        try (Connection connection = createConnection()) {
            //1.开启信道
            Channel channel = connection.createChannel();
            //2.设置Confirm模式
            channel.confirmSelect();
            //3.声明队列
            channel.queueDeclare("queue2",true,false,false,null);
            //4.发送消息，等待确认
            //当消息数量达到一定的时候就进行确认
            long start = System.currentTimeMillis();
            int batchSize = 100;
            int outstandingMessageCount = 0;
            for (int i = 0; i < 1000; i++) {
                String message = "hello publisher Confirms  " +  i;
                channel.basicPublish("","queue2",null,message.getBytes());
                outstandingMessageCount++;
                //等待确认
                if (outstandingMessageCount == batchSize) {
                    channel.waitForConfirms(5000);
                    outstandingMessageCount = 0;
                }
            }
            if (outstandingMessageCount > 0) {
                channel.waitForConfirms(5000);
            }
            long end = System.currentTimeMillis();
            System.out.println("批量确认策略消息条数: " + 1000 + " 耗时: " + (end-start));
        }
        
    }

    private static void publishingMessagesIndividually() throws IOException, TimeoutException, InterruptedException {
        try (Connection connection = createConnection()) {
            //1.开启通信
            Channel channel = connection.createChannel();
            //2.设置Confirm模式
            channel.confirmSelect();
            //3.声明队列
            channel.queueDeclare("queue1",true,false,false,null);
            long start = System.currentTimeMillis();
            //4.发送消息，等待确认
            for (int i = 0; i < 1000; i++) {
                String message = "hello publisher Confirms  " +  i;
                channel.basicPublish("","queue1",null,message.getBytes());
                //等待确认
                channel.waitForConfirms(5000);
            }
            long end = System.currentTimeMillis();
            System.out.println("单独确认策略消息条数: " + 1000 + " 耗时: " + (end-start));
        }

    }
}
